//taillog 依赖于 github.com/hpcloud/tail
package taillog

import (
	"context"
	"fmt"
	"github.com/MingOf/logDemo/kafka"
	"github.com/Shopify/sarama"
	"github.com/hpcloud/tail"
	"logagent/etcd"
	"time"
)

//新的结构体Tail,这个结构体是 tail.Tail类型的扩展，包含有 TailContext 和一个 *tail.Tail类型的结构体
type Tail struct {
	Tl          *tail.Tail
	TailContext *TailContext
	filePath    string
	Topic       string
	Producer    sarama.SyncProducer
}

// 结构体，用于保存Context
type TailContext struct {
	Ctx    context.Context
	Cancel context.CancelFunc
}

var Tails = make(map[string]*Tail)

//新建一个Tail
//
//参数：
//	tc *TailContext,
//	filePath string
//返回：
//	*Tail,
//	error
func NewTail(tc *TailContext, filePath string, topic string) (*Tail, error) {
	// tailf 用法
	//filePath = "./my.log"
	config := tail.Config{
		ReOpen:    true,
		Follow:    true,
		Location:  &tail.SeekInfo{Offset: 0, Whence: 2},
		MustExist: false,
		Poll:      true,
	}
	tl, err := tail.TailFile(filePath, config)
	if err != nil {
		fmt.Println("[pkg taillog] tail file failed, err:", err)
		return nil, err
	}
	newTail := &Tail{
		Tl:          tl,
		TailContext: tc,
		filePath:    filePath,
		Topic:       topic,
	}
	//将新建的Tail加入map，该map使用filePath作为key
	Tails[filePath] = newTail
	return newTail, nil
}

//获取 Tail.tl.Lines
func (t *Tail) GetLines() <-chan *tail.Line {
	select {
	case <-t.TailContext.Ctx.Done():
		t.Stop()
		return nil
	default:
		return t.Tl.Lines
	}
}

// 取消对应的 context
//
// 停止对文件的 tail 操作
//
//从 Tails 中删除对应的 Tail
func (t *Tail) Stop() {
	t.TailContext.Cancel()
	if err := t.Tl.Stop(); err != nil {
		fmt.Println("[pkg taillog] stop tail failed,err: ", err)
	}
	delete(Tails, t.filePath)
	fmt.Println("[pkg taillog] tail is stopped,path: ", t.filePath)
}

/*
根据传入的参数对 Tails 进行增减
*/
func Diff(eventType string, minus map[int]*etcd.LogEntry, plus map[int]*etcd.LogEntry, producer sarama.SyncProducer) {

	//如果是 del 操作则停车tail并将清空Tails这个map结构
	if eventType == "DELETE" {
		for _, t := range Tails {
			t.Stop()
		}
		Tails = make(map[string]*Tail)
	}
	//如果put操作后的值和相比有增加则新增一个Tail
	for _, entry := range plus {
		if Tails[entry.Path] == nil {
			fmt.Println("[pkg taillog] new Path:", entry.Path)
			ctx, cancel := context.WithCancel(context.Background())
			newTail, err := NewTail(&TailContext{Ctx: ctx, Cancel: cancel}, entry.Path, entry.Topic)
			if err != nil {
				fmt.Println("[pkg taillog] NewTail err:", err)
			}
			go newTail.SendToKafka(producer)
		}
	}
	//如果减少则剔除(调用Stop方法后自动删除)
	for _, entry := range minus {
		if Tails[entry.Path] != nil {
			Tails[entry.Path].Stop()
		}
	}
}

/*
发送到kafka ，这个函数是 kafka.SendToKafka 的封装
*/
func (t *Tail) SendToKafka(producer sarama.SyncProducer) {
	lines := t.GetLines()
LOOP:
	for {
		select {
		case <-t.TailContext.Ctx.Done():
			//wg.Done()
			t.Stop()
			fmt.Printf("[pkg taillog] sendToKafka canceled,topic:%s, path:%s\n", t.Topic, t.filePath)
			break LOOP
		case line := <-lines:
			fmt.Println(line.Text)
			kafka.SendToKafka(producer, t.Topic, line.Text)
		default:
			time.Sleep(time.Second)
		}
	}

}
